package com.wanqian.redis.activemq;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * @Author: wanqian
 * @Date: 2021/3/13 13:34
 */
public class ActiveMQConsumer {

    private static final String URL = "tcp://145.170.39.203:61616";

    private static final String QUEUE_NAME = "queue_test";

    private static final String TOPIC_NAME = "topic_test";

    public static void main(String[] args) {
        //创建连接工厂
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(URL);
//        TopicSubscriber subscriber;
        MessageConsumer consumer;
        Connection connection;
        //创建连接和会话(设置事务和签收)
        try {
            connection = connectionFactory.createConnection();
            connection.setClientID("testConsumer");
            Session session = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
            //创建队列
//            Queue queue = session.createQueue(QUEUE_NAME);
            Topic topic = session.createTopic(TOPIC_NAME);
            //创建消费者(指定destination)
             consumer = session.createConsumer(topic);
//            subscriber = session.createDurableSubscriber(topic, "test...");

            //同步阻塞接收
//            Message message = consumer.receive();
//            while (message != null) {
//                TextMessage textMessage = (TextMessage) message;
//                System.out.println(textMessage.getText());
//                message = consumer.receive(2000);
//            }
            //开启连接
            connection.start();
            //异步非阻塞接收
            consumer.setMessageListener(message -> {
                if (message != null) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println(textMessage.getText());
                        //手动签收
                        textMessage.acknowledge();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            // 提交事务
            // 事务大于签收，提交事务时自动全部签收
            // 开启事务后必须commit，手动签收无效
            session.commit();

            consumer.close();
            session.close();
            connection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
